package com.flute.haflute.jobbox.base.storage;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import me.prettyprint.cassandra.service.CassandraClient;
import me.prettyprint.cassandra.service.CassandraClientPool;
import me.prettyprint.cassandra.service.CassandraClientPoolFactory;
import me.prettyprint.cassandra.service.Keyspace;
import me.prettyprint.cassandra.service.PoolExhaustedException;

import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.jobbox.base.exception.ServiceUnavailableException;

public class CassandraStorageServicePool implements StorageServicePool {
	private static Logger logger = LoggerFactory
			.getLogger(CassandraStorageServicePool.class);

	private static CassandraClientPool clientPool;

	private Properties prop;

	private List<CassandraStorageServiceImpl> servicePool;

	private CassandraClient client;

	private String[] cassandraHosts = {};

	private boolean localCassandraFirst = false;

	private final String COLUMNFAMILY = "cassandra.Keyspace.ColumnFamily";

	protected CassandraStorageServicePool(Properties props) throws Exception {
		initialCassandraEnvironment(props);
	}

	/**
	 * @throws Exception
	 * @throws UnavailableException
	 */
	private synchronized void initialCassandraEnvironment(Properties props)
			throws Exception {
		clientPool = CassandraClientPoolFactory.INSTANCE.get();
		try {
			this.prop = props;
			String hosts = prop.getProperty("cassandra.hosts", "localhost");
			String port = prop.getProperty("cassandra.ThriftPort", "9160");
			String localAccessFirst = prop.getProperty(
					"cassandra.localaccess.first", "true");
			if ("true".equalsIgnoreCase(localAccessFirst)) {
				localCassandraFirst = true;
				hosts = hosts.replaceAll("localhost", "");
				hosts = hosts.replaceAll(";;", ";");
			}
			if (hosts.indexOf(';') == 0) {
				hosts = hosts.substring(1);
			}
			String[] hostArray = hosts.split(";");
			cassandraHosts = new String[hostArray.length];
			for (int i = 0; i < hostArray.length; i++) {
				cassandraHosts[i] = hostArray[i] + ":" + port;
			}
		} catch (Exception e) {
			logger.error("will set cassandra to localhost, because error", e);
		}

		// 30 = 50 * 0.6, 50 is the max client of hector
		servicePool = new ArrayList<CassandraStorageServiceImpl>(
				30 * cassandraHosts.length);
		installConnection();
	}

	/**
	 * @throws Exception
	 */
	private void installConnection() throws Exception {
		try {
			if (localCassandraFirst) {
				client = clientPool.borrowClient("localhost:"
						+ prop.getProperty("cassandra.ThriftPort", "9160"));
			}
		} catch (Exception e) {
			logger.info("can't connect to localhost Cassandra", e);
			client = null;
			localCassandraFirst = false;
		}

		if (client == null)
			client = clientPool.borrowClient(cassandraHosts);

		addService2ServicePool();
	}

	/**
	 * @throws TException
	 * @throws NotFoundException
	 * @throws IllegalArgumentException
	 */
	private synchronized CassandraStorageServiceImpl addService2ServicePool()
			throws IllegalArgumentException, NotFoundException, TException {
		Keyspace keyspace = client.getKeyspace(prop.getProperty(
				"cassandra.Keyspace", "Keyspace1"));
		ColumnPath columnPath = new ColumnPath(prop.getProperty(COLUMNFAMILY,
				"Standard1"));
		CassandraStorageServiceImpl service = new CassandraStorageServiceImpl(
				keyspace, columnPath, localCassandraFirst);
		servicePool.add(service);
		return service;
	}

	protected synchronized void release() {
		for (CassandraStorageServiceImpl service : servicePool) {
			service.releaseClient(clientPool);
		}
	}

	@Override
	public synchronized KeyValueStorageService borrowService()
			throws ServiceUnavailableException {
		for (CassandraStorageServiceImpl service : servicePool) {
			if (service.canBorrow()) {
				service.markAsBorrowed();
				return service;
			}
		}

		try {
			if (localCassandraFirst && servicePool.size() < 50) {
				client = clientPool.borrowClient();
			} else {
				client = clientPool.borrowClient(cassandraHosts);
			}
			CassandraStorageServiceImpl service = addService2ServicePool();
			service.markAsBorrowed();
			return service;
		} catch (PoolExhaustedException e) {
			logger.info("Cassandra Pool exhausted, please wait", e);
			throw new ServiceUnavailableException(e.getStackTrace());
		} catch (Exception e) {
			logger.error("can not get more Cassandra Client", e);
			throw new ServiceUnavailableException(e.getStackTrace());
		}
	}

	@Override
	public synchronized void releaseService(KeyValueStorageService service) {
		if (service instanceof CassandraStorageServiceImpl) {
			((CassandraStorageServiceImpl) service).markAsUnborrowed();
		}
	}
}
